package mapReduce.demo07_join.demo01_productOrderReduceJoin;

import mapReduce.util.JobSubmit;
import mapReduce.writable.ProductOrderWritable;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class ProductOrderReduceJoinApp {

    public static class POJMapper extends Mapper<LongWritable, Text, ProductOrderWritable, NullWritable> {

        ProductOrderWritable outKey = new ProductOrderWritable();

        @Override
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, ProductOrderWritable, NullWritable>.Context context) throws IOException, InterruptedException {

            String line = value.toString();
            String[] fields = line.split("\t");

            //如果拆分后的字段数量为3，就是order数据
            if(fields.length==3){
                outKey.setOrderId(fields[0]);
                outKey.setProductId(fields[1]);
                outKey.setProductCount(Integer.parseInt(fields[2]));
                outKey.setFlag("order");
            }else if(fields.length==2){
                //如果拆分后的字段数量为2，就是product数据
                outKey.setProductId(fields[0]);
                outKey.setProductName(fields[1]);
                outKey.setFlag("product");
            }

            context.write(outKey,NullWritable.get());
        }
    }

    public static class POJReducer extends Reducer<ProductOrderWritable,NullWritable,Text,NullWritable>{

        Text outKey = new Text();

        @Override
        protected void reduce(ProductOrderWritable key, Iterable<NullWritable> values, Reducer<ProductOrderWritable, NullWritable, Text, NullWritable>.Context context) throws IOException, InterruptedException {

            /**
             * 接收到的数据格式,在一个组中(reduce方法的一次执行)订单数据可以有多条，但是商品数据只能有唯一一条
             *      1001    01  5
             *      1002    01  8
             *      01  辣条
             *      1003    01  10
             *      ....
             */
            //初始化一个订单数据的集合
            List<ProductOrderWritable> orderList = new ArrayList<>();
            //初始化商品名称，用来接收此组内唯一的商品数据
            String productName = "";

            //遍历values，按照value找位置，用的是key的数据
            for (NullWritable value : values) {
                String flag = key.getFlag();
                if(flag.equals("order")){
                    try {
                        ProductOrderWritable destOrder = new ProductOrderWritable();
                        BeanUtils.copyProperties(destOrder,key);
                        orderList.add(destOrder);
                    }catch (Exception e){
                        e.printStackTrace();
                    }

                }else if(flag.equals("product")){
                    productName = key.getProductName();
                }
            }

            for (ProductOrderWritable order : orderList) {
                order.setProductName(productName);
                outKey.set(order.toString());
                context.write(outKey,NullWritable.get());
            }
        }
    }

    public static void main(String[] args) {
        JobSubmit.submitBaseJob(
                ProductOrderReduceJoinApp.class,
                args,
                "group",
                ProductOrderGroupingComparator.class
        );
    }

}
